/*
 *
 * Copyright 2021 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package xdsresource

import (
	"fmt"
	v3clusterpb "github.com/envoyproxy/go-control-plane/envoy/config/cluster/v3"
	v3corepb "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
	"google.golang.org/protobuf/types/known/structpb"
	"net"
	"strconv"
	"strings"
	"time"
	"tmlake.com/api-gateway/istio-agent/pkg/pretty"

	//v3aggregateclusterpb "github.com/envoyproxy/go-control-plane/envoy/extensions/clusters/aggregate/v3"
	v3tlspb "github.com/envoyproxy/go-control-plane/envoy/extensions/transport_sockets/tls/v3"
	"github.com/golang/protobuf/proto"
	"google.golang.org/protobuf/types/known/anypb"
	"tmlake.com/api-gateway/istio-agent/pkg/envconfig"
	"tmlake.com/api-gateway/istio-agent/pkg/grpclog/inside"
	"tmlake.com/api-gateway/istio-agent/xds/internal/matcher"
	"tmlake.com/api-gateway/istio-agent/xds/xdsclient/xdsresource/version"
)

// TransportSocket proto message has a `name` field which is expected to be set
// to this value by the management server.
const transportSocketName = "envoy.transport_sockets.tls"

// UnmarshalCluster processes resources received in an CDS response, validates
// them, and transforms them into a native struct which contains only fields we
// are interested in.
func UnmarshalCluster(opts *UnmarshalOptions) (map[string]ClusterUpdateErrTuple, UpdateMetadata, error) {
	update := make(map[string]ClusterUpdateErrTuple)
	md, err := processAllResources(opts, update)
	return update, md, err
}

func unmarshalClusterResource(r *anypb.Any, f UpdateValidatorFunc, logger *grpclog.PrefixLogger) (string, ClusterUpdate, error) {
	r, err := unwrapResource(r)
	if err != nil {
		return "", ClusterUpdate{}, fmt.Errorf("failed to unwrap resource: %v", err)
	}

	if !IsClusterResource(r.GetTypeUrl()) {
		return "", ClusterUpdate{}, fmt.Errorf("unexpected resource type: %q ", r.GetTypeUrl())
	}

	cluster := &v3clusterpb.Cluster{}
	if err := proto.Unmarshal(r.GetValue(), cluster); err != nil {
		return "", ClusterUpdate{}, fmt.Errorf("failed to unmarshal resource: %v", err)
	}
	logger.Debugf("Resource with name: %v, type: %T, contains: %v", cluster.GetName(), cluster, pretty.ToJSON(cluster))
	logger.Infof("Resource with name: %v, type: %T", cluster.GetName(), cluster)

	cu, err := validateClusterAndConstructClusterUpdate(cluster)
	if err != nil {
		return cluster.GetName(), ClusterUpdate{}, err
	}

	cu.Raw = r
	if f != nil {
		if err := f(cu); err != nil {
			return "", ClusterUpdate{}, err
		}
	}

	return cluster.GetName(), cu, nil
}

const (
	defaultRingHashMinSize = 1024
	defaultRingHashMaxSize = 8 * 1024 * 1024 // 8M
	ringHashSizeUpperBound = 8 * 1024 * 1024 // 8M
)

func validateClusterAndConstructClusterUpdate(cluster *v3clusterpb.Cluster) (ClusterUpdate, error) {
	var lbPolicy *ClusterLBPolicyRingHash
	switch cluster.GetLbPolicy() {
	case v3clusterpb.Cluster_ROUND_ROBIN:
		lbPolicy = nil // The default is round_robin, and there's no config to set.
	case v3clusterpb.Cluster_RING_HASH:
		rhc := cluster.GetRingHashLbConfig()
		if rhc.GetHashFunction() != v3clusterpb.Cluster_RingHashLbConfig_XX_HASH {
			return ClusterUpdate{}, fmt.Errorf("unsupported ring_hash hash function %v in response: %+v", rhc.GetHashFunction(), cluster)
		}
		// Minimum defaults to 1024 entries, and limited to 8M entries Maximum
		// defaults to 8M entries, and limited to 8M entries
		var minSize, maxSize uint64 = defaultRingHashMinSize, defaultRingHashMaxSize
		if min := rhc.GetMinimumRingSize(); min != nil {
			if min.GetValue() > ringHashSizeUpperBound {
				return ClusterUpdate{}, fmt.Errorf("unexpected ring_hash mininum ring size %v in response: %+v", min.GetValue(), cluster)
			}
			minSize = min.GetValue()
		}
		if max := rhc.GetMaximumRingSize(); max != nil {
			if max.GetValue() > ringHashSizeUpperBound {
				return ClusterUpdate{}, fmt.Errorf("unexpected ring_hash maxinum ring size %v in response: %+v", max.GetValue(), cluster)
			}
			maxSize = max.GetValue()
		}
		if minSize > maxSize {
			return ClusterUpdate{}, fmt.Errorf("ring_hash config min size %v is greater than max %v", minSize, maxSize)
		}
		lbPolicy = &ClusterLBPolicyRingHash{MinimumRingSize: minSize, MaximumRingSize: maxSize}
	default:
		lbPolicy = nil
	}

	// Process security configuration received from the control plane
	sc, err := securityConfigFromCluster(cluster)
	if err != nil {
		return ClusterUpdate{}, err
	}

	if sc != nil {
		xdsresourceLogger.Infof(printSecurityConfig(sc, cluster.GetName()))
	}
	// Process outlier detection received from the control plane iff the
	// corresponding environment variable is set.
	var od *OutlierDetection
	if envconfig.XDSOutlierDetection {
		var err error
		if od, err = outlierConfigFromCluster(cluster); err != nil {
			return ClusterUpdate{}, err
		}
	}

	ret := ClusterUpdate{
		ClusterName:      cluster.GetName(),
		SecurityCfg:      sc,
		MaxRequests:      circuitBreakersFromCluster(cluster),
		LBPolicy:         lbPolicy,
		OutlierDetection: od,
	}

	// Note that this is different from the gRFC (gRFC A47 says to include the
	// full ServerConfig{URL,creds,server feature} here). This information is
	// not available here, because this function doesn't have access to the
	// xdsclient bootstrap information now (can be added if necessary). The
	// ServerConfig will be read and populated by the CDS balancer when
	// processing this field.
	if cluster.GetLrsServer().GetSelf() != nil {
		ret.LRSServerConfig = ClusterLRSServerSelf
	}

	// Validate and set cluster type from the response.
	switch {
	case cluster.GetType() == v3clusterpb.Cluster_EDS:
		if configsource := cluster.GetEdsClusterConfig().GetEdsConfig(); configsource.GetAds() == nil && configsource.GetSelf() == nil {
			return ClusterUpdate{}, fmt.Errorf("CDS's EDS config source is not ADS or Self: %+v", cluster)
		}
		ret.ClusterType = ClusterTypeEDS
		ret.EDSServiceName = cluster.GetEdsClusterConfig().GetServiceName()
		return ret, nil
	default:
		return ClusterUpdate{}, nil
	}
}

// dnsHostNameFromCluster extracts the DNS host name from the cluster's load
// assignment.
//
// There should be exactly one locality, with one endpoint, whose address
// contains the address and port.
func dnsHostNameFromCluster(cluster *v3clusterpb.Cluster) (string, error) {
	loadAssignment := cluster.GetLoadAssignment()
	if loadAssignment == nil {
		return "", fmt.Errorf("load_assignment not present for LOGICAL_DNS cluster")
	}
	if len(loadAssignment.GetEndpoints()) != 1 {
		return "", fmt.Errorf("load_assignment for LOGICAL_DNS cluster must have exactly one locality, got: %+v", loadAssignment)
	}
	endpoints := loadAssignment.GetEndpoints()[0].GetLbEndpoints()
	if len(endpoints) != 1 {
		return "", fmt.Errorf("locality for LOGICAL_DNS cluster must have exactly one endpoint, got: %+v", endpoints)
	}
	endpoint := endpoints[0].GetEndpoint()
	if endpoint == nil {
		return "", fmt.Errorf("endpoint for LOGICAL_DNS cluster not set")
	}
	socketAddr := endpoint.GetAddress().GetSocketAddress()
	if socketAddr == nil {
		return "", fmt.Errorf("socket address for endpoint for LOGICAL_DNS cluster not set")
	}
	if socketAddr.GetResolverName() != "" {
		return "", fmt.Errorf("socket address for endpoint for LOGICAL_DNS cluster not set has unexpected custom resolver name: %v", socketAddr.GetResolverName())
	}
	host := socketAddr.GetAddress()
	if host == "" {
		return "", fmt.Errorf("host for endpoint for LOGICAL_DNS cluster not set")
	}
	port := socketAddr.GetPortValue()
	if port == 0 {
		return "", fmt.Errorf("port for endpoint for LOGICAL_DNS cluster not set")
	}
	return net.JoinHostPort(host, strconv.Itoa(int(port))), nil
}

func securityConfigFromTransportSocket(ts *v3corepb.TransportSocket) (*SecurityConfig, error) {
	if name := ts.GetName(); name != transportSocketName {
		return nil, fmt.Errorf("transport_socket field has unexpected name: %s", name)
	}
	any := ts.GetTypedConfig()
	if any == nil || any.TypeUrl != version.V3UpstreamTLSContextURL {
		return nil, fmt.Errorf("transport_socket field has unexpected typeURL: %s", any.TypeUrl)
	}
	upstreamCtx := &v3tlspb.UpstreamTlsContext{}
	if err := proto.Unmarshal(any.GetValue(), upstreamCtx); err != nil {
		return nil, fmt.Errorf("failed to unmarshal UpstreamTlsContext in CDS response: %v", err)
	}
	// The following fields from `UpstreamTlsContext` are ignored:
	// - sni
	// - allow_renegotiation
	// - max_session_keys
	if upstreamCtx.GetCommonTlsContext() == nil {
		return nil, nil
	}

	return securityConfigFromCommonTLSContext(upstreamCtx.GetCommonTlsContext(), false)
}

// securityConfigFromCluster extracts the relevant security configuration from
// the received Cluster resource.
func securityConfigFromCluster(cluster *v3clusterpb.Cluster) (*SecurityConfig, error) {
	var transportSocket *v3corepb.TransportSocket
	//We only deal with first tlsMode: istio matche
	if tsm := cluster.GetTransportSocketMatches(); len(tsm) != 0 {
		isMatch := false
		for _, tm := range tsm {
			if tm.GetMatch() != nil {
				for k, v := range tm.GetMatch().Fields {
					if k == TransportSocketMatchName {
						switch v.GetKind().(type) {
						case *structpb.Value_StringValue:
							if v.GetStringValue() == TransportSocketMatchValue {
								transportSocket = tm.GetTransportSocket()
							}
							isMatch = true
							break
						}
					}
				}
			}
			if isMatch {
				break
			}
		}
		if transportSocket == nil {
			return nil, nil
		}
		return securityConfigFromTransportSocket(transportSocket)
	}
	// The Cluster resource contains a `transport_socket` field, which contains
	// a oneof `typed_config` field of type `protobuf.Any`. The any proto
	// contains a marshaled representation of an `UpstreamTlsContext` message.
	ts := cluster.GetTransportSocket()
	if ts == nil {
		return nil, nil
	}
	return securityConfigFromTransportSocket(ts)
}

// common is expected to be not nil.
// The `alpn_protocols` field is ignored.
func securityConfigFromCommonTLSContext(common *v3tlspb.CommonTlsContext, server bool) (*SecurityConfig, error) {
	if common.GetCustomHandshaker() != nil {
		return nil, fmt.Errorf("unsupported custom_handshaker field in CommonTlsContext message: %+v", common)
	}

	sc := &SecurityConfig{}
	switch t := common.GetValidationContextType().(type) {
	case *v3tlspb.CommonTlsContext_ValidationContext:
		validationContext := common.GetValidationContext()
		if validationContext == nil {
			return nil, fmt.Errorf("validation context is nil")
		}
		var matchers []matcher.StringMatcher
		for _, m := range validationContext.GetMatchSubjectAltNames() {
			matcher, err := matcher.StringMatcherFromProto(m)
			if err != nil {
				return nil, err
			}
			matchers = append(matchers, matcher)
		}

		sc.CommonTlsContext.ValidationContext.SubjectAltNameMatchers = matchers
	case *v3tlspb.CommonTlsContext_ValidationContextSdsSecretConfig:
		validationContextSdsSecretConfig := common.GetValidationContextSdsSecretConfig()
		if validationContextSdsSecretConfig == nil {
			sc.CommonTlsContext.ValidationContextSdsSecretConfig = nil
			return nil, fmt.Errorf("validation context SdsSecretConfig is nil")
		}
		sc.CommonTlsContext.ValidationContextSdsSecretConfig = &SdsSecretConfig{}

		sc.CommonTlsContext.ValidationContextSdsSecretConfig.Name = validationContextSdsSecretConfig.Name
		if validationContextSdsSecretConfig.GetSdsConfig() != nil {
			sc.CommonTlsContext.ValidationContextSdsSecretConfig.IsSDS = true
		}

	case *v3tlspb.CommonTlsContext_CombinedValidationContext:
		combined := common.GetCombinedValidationContext()
		if combined == nil {
			return nil, fmt.Errorf("Combined validation context is nil")
		}

		var matchers []matcher.StringMatcher
		if def := combined.GetDefaultValidationContext(); def != nil {
			for _, m := range def.GetMatchSubjectAltNames() {
				matcher, err := matcher.StringMatcherFromProto(m)
				if err != nil {
					return nil, err
				}
				matchers = append(matchers, matcher)
			}
		}

		sc.CommonTlsContext.ValidationContext.SubjectAltNameMatchers = matchers

		if sdsSecretConfig := combined.GetValidationContextSdsSecretConfig(); sdsSecretConfig != nil {
			sc.CommonTlsContext.ValidationContextSdsSecretConfig = &SdsSecretConfig{}
			sc.CommonTlsContext.ValidationContextSdsSecretConfig.Name = sdsSecretConfig.Name
			if sdsSecretConfig.GetSdsConfig() != nil {
				sc.CommonTlsContext.ValidationContextSdsSecretConfig.IsSDS = true
			}
		} else {
			sc.CommonTlsContext.ValidationContextSdsSecretConfig = nil
		}
	case nil:
		return nil, fmt.Errorf("validation context validation context type is nil")
	default:
		return nil, fmt.Errorf("validation context contains unexpected type: %T", t)
	}

	if tlsCertificateSdsSecretConfigs := common.GetTlsCertificateSdsSecretConfigs(); tlsCertificateSdsSecretConfigs != nil {
		for _, sdsSecretConfig := range tlsCertificateSdsSecretConfigs {
			if sdsSecretConfig != nil {
				var sds SdsSecretConfig
				sds.Name = sdsSecretConfig.Name
				if sdsSecretConfig.GetSdsConfig() != nil {
					sds.IsSDS = true
				}

				sc.CommonTlsContext.TlsCertificateSdsSecretConfigs = append(sc.CommonTlsContext.TlsCertificateSdsSecretConfigs, &sds)
			}
		}
	}

	return sc, nil
}

func printSecurityConfig(sc *SecurityConfig, clusterName string) string {
	if sc == nil {
		return ""
	}

	var commonTlsContext strings.Builder
	commonTlsContext.WriteString(fmt.Sprintf("clusterName:%q, CommonTlsContext: %q:%v, ", clusterName, sc.CommonTlsContext.ValidationContextSdsSecretConfig.Name, sc.CommonTlsContext.ValidationContextSdsSecretConfig.IsSDS))
	if len(sc.CommonTlsContext.ValidationContext.SubjectAltNameMatchers) > 0 {
		commonTlsContext.WriteString("SubjectAltNameMatchers: ")
		for i, subjectAltNameMatcher := range sc.CommonTlsContext.ValidationContext.SubjectAltNameMatchers {
			if i != 0 {
				commonTlsContext.WriteString(";\n")
			}

			commonTlsContext.WriteString(fmt.Sprintf("mode %q: %q, ", subjectAltNameMatcher.GetMode(), subjectAltNameMatcher.GetMatcher()))

		}
	}
	if len(sc.CommonTlsContext.TlsCertificateSdsSecretConfigs) > 0 {
		var i int
		for _, tlsCertificateSdsSecretConfig := range sc.CommonTlsContext.TlsCertificateSdsSecretConfigs {
			if i != 0 {
				commonTlsContext.WriteString(";\n")
			}
			i++
			commonTlsContext.WriteString(fmt.Sprintf("TlsCertificateSdsSecretConfig: %q: %v", tlsCertificateSdsSecretConfig.Name, tlsCertificateSdsSecretConfig.IsSDS))
		}
	}
	return commonTlsContext.String()
}

// circuitBreakersFromCluster extracts the circuit breakers configuration from
// the received cluster resource. Returns nil if no CircuitBreakers or no
// Thresholds in CircuitBreakers.
func circuitBreakersFromCluster(cluster *v3clusterpb.Cluster) *uint32 {
	for _, threshold := range cluster.GetCircuitBreakers().GetThresholds() {
		if threshold.GetPriority() != v3corepb.RoutingPriority_DEFAULT {
			continue
		}
		maxRequestsPb := threshold.GetMaxRequests()
		if maxRequestsPb == nil {
			return nil
		}
		maxRequests := maxRequestsPb.GetValue()
		return &maxRequests
	}
	return nil
}

// outlierConfigFromCluster extracts the relevant outlier detection
// configuration from the received cluster resource. Returns nil if no
// OutlierDetection field set in the cluster resource.
func outlierConfigFromCluster(cluster *v3clusterpb.Cluster) (*OutlierDetection, error) {
	od := cluster.GetOutlierDetection()
	if od == nil {
		return nil, nil
	}
	const (
		defaultInterval                       = 10 * time.Second
		defaultBaseEjectionTime               = 30 * time.Second
		defaultMaxEjectionTime                = 300 * time.Second
		defaultMaxEjectionPercent             = 10
		defaultSuccessRateStdevFactor         = 1900
		defaultEnforcingSuccessRate           = 100
		defaultSuccessRateMinimumHosts        = 5
		defaultSuccessRateRequestVolume       = 100
		defaultFailurePercentageThreshold     = 85
		defaultEnforcingFailurePercentage     = 0
		defaultFailurePercentageMinimumHosts  = 5
		defaultFailurePercentageRequestVolume = 50
	)
	// "The google.protobuf.Duration fields interval, base_ejection_time, and
	// max_ejection_time must obey the restrictions in the
	// google.protobuf.Duration documentation and they must have non-negative
	// values." - A50
	interval := defaultInterval
	if i := od.GetInterval(); i != nil {
		if err := i.CheckValid(); err != nil {
			return nil, fmt.Errorf("outlier_detection.interval is invalid with error %v", err)
		}
		if interval = i.AsDuration(); interval < 0 {
			return nil, fmt.Errorf("outlier_detection.interval = %v; must be a valid duration and >= 0", interval)
		}
	}

	baseEjectionTime := defaultBaseEjectionTime
	if bet := od.GetBaseEjectionTime(); bet != nil {
		if err := bet.CheckValid(); err != nil {
			return nil, fmt.Errorf("outlier_detection.base_ejection_time is invalid with error %v", err)
		}
		if baseEjectionTime = bet.AsDuration(); baseEjectionTime < 0 {
			return nil, fmt.Errorf("outlier_detection.base_ejection_time = %v; must be >= 0", baseEjectionTime)
		}
	}

	maxEjectionTime := defaultMaxEjectionTime
	if met := od.GetMaxEjectionTime(); met != nil {
		if err := met.CheckValid(); err != nil {
			return nil, fmt.Errorf("outlier_detection.max_ejection_time is invalid with error %v", err)
		}
		if maxEjectionTime = met.AsDuration(); maxEjectionTime < 0 {
			return nil, fmt.Errorf("outlier_detection.max_ejection_time = %v; must be >= 0", maxEjectionTime)
		}
	}

	// "The fields max_ejection_percent, enforcing_success_rate,
	// failure_percentage_threshold, and enforcing_failure_percentage must have
	// values less than or equal to 100. If any of these requirements is
	// violated, the Cluster resource should be NACKed." - A50
	maxEjectionPercent := uint32(defaultMaxEjectionPercent)
	if mep := od.GetMaxEjectionPercent(); mep != nil {
		if maxEjectionPercent = mep.GetValue(); maxEjectionPercent > 100 {
			return nil, fmt.Errorf("outlier_detection.max_ejection_percent = %v; must be <= 100", maxEjectionPercent)
		}
	}
	enforcingSuccessRate := uint32(defaultEnforcingSuccessRate)
	if esr := od.GetEnforcingSuccessRate(); esr != nil {
		if enforcingSuccessRate = esr.GetValue(); enforcingSuccessRate > 100 {
			return nil, fmt.Errorf("outlier_detection.enforcing_success_rate = %v; must be <= 100", enforcingSuccessRate)
		}
	}
	failurePercentageThreshold := uint32(defaultFailurePercentageThreshold)
	if fpt := od.GetFailurePercentageThreshold(); fpt != nil {
		if failurePercentageThreshold = fpt.GetValue(); failurePercentageThreshold > 100 {
			return nil, fmt.Errorf("outlier_detection.failure_percentage_threshold = %v; must be <= 100", failurePercentageThreshold)
		}
	}
	enforcingFailurePercentage := uint32(defaultEnforcingFailurePercentage)
	if efp := od.GetEnforcingFailurePercentage(); efp != nil {
		if enforcingFailurePercentage = efp.GetValue(); enforcingFailurePercentage > 100 {
			return nil, fmt.Errorf("outlier_detection.enforcing_failure_percentage = %v; must be <= 100", enforcingFailurePercentage)
		}
	}

	successRateStdevFactor := uint32(defaultSuccessRateStdevFactor)
	if srsf := od.GetSuccessRateStdevFactor(); srsf != nil {
		successRateStdevFactor = srsf.GetValue()
	}
	successRateMinimumHosts := uint32(defaultSuccessRateMinimumHosts)
	if srmh := od.GetSuccessRateMinimumHosts(); srmh != nil {
		successRateMinimumHosts = srmh.GetValue()
	}
	successRateRequestVolume := uint32(defaultSuccessRateRequestVolume)
	if srrv := od.GetSuccessRateRequestVolume(); srrv != nil {
		successRateRequestVolume = srrv.GetValue()
	}
	failurePercentageMinimumHosts := uint32(defaultFailurePercentageMinimumHosts)
	if fpmh := od.GetFailurePercentageMinimumHosts(); fpmh != nil {
		failurePercentageMinimumHosts = fpmh.GetValue()
	}
	failurePercentageRequestVolume := uint32(defaultFailurePercentageRequestVolume)
	if fprv := od.GetFailurePercentageRequestVolume(); fprv != nil {
		failurePercentageRequestVolume = fprv.GetValue()
	}

	return &OutlierDetection{
		Interval:                       interval,
		BaseEjectionTime:               baseEjectionTime,
		MaxEjectionTime:                maxEjectionTime,
		MaxEjectionPercent:             maxEjectionPercent,
		EnforcingSuccessRate:           enforcingSuccessRate,
		FailurePercentageThreshold:     failurePercentageThreshold,
		EnforcingFailurePercentage:     enforcingFailurePercentage,
		SuccessRateStdevFactor:         successRateStdevFactor,
		SuccessRateMinimumHosts:        successRateMinimumHosts,
		SuccessRateRequestVolume:       successRateRequestVolume,
		FailurePercentageMinimumHosts:  failurePercentageMinimumHosts,
		FailurePercentageRequestVolume: failurePercentageRequestVolume,
	}, nil
}
